package org.sunrise.game.game.message;

import com.alibaba.fastjson2.JSON;
import org.sunrise.game.common.log.LogCore;
import org.sunrise.game.common.message.GameMessage;
import org.sunrise.game.common.message.MessageUtils;
import org.sunrise.game.common.network.message.ClientMessageManager;
import org.sunrise.game.game.async.AsyncEventManager;
import org.sunrise.game.game.db.DbManager;
import org.sunrise.game.game.human.HumanObject;
import org.sunrise.game.game.human.HumanObjectManger;
import org.sunrise.game.game.logic.LogicUtils;
import org.sunrise.game.game.logic.ProtoParserUtils;
import org.sunrise.game.game.logic.ToolsUtils;
import org.sunrise.game.game.logic.system.GameSystem;
import org.sunrise.game.game.login.LoginMsgHandler;
import org.sunrise.game.game.modules.MapModule;
import org.sunrise.game.game.proto.TopicProto;
import org.sunrise.game.rpc.function.Call;
import org.sunrise.game.rpc.function.CallType;
import org.sunrise.game.rpc.function.RpcFunction;
import org.sunrise.game.rpc.function.RpcManager;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * game消息处理类，来自对外服和跨服的消息，全部在此处理
 */
public class GameMasterMessageManager extends ClientMessageManager {
    public GameMasterMessageManager(String nodeId) {
        super(nodeId);
    }

    @Override
    public void pulse() {
        try {
            // 处理消息 放入玩家自身队列
            pulseHandler();
            // 处理玩家队列中的消息
            pulseHandlerHumanMsg();
            // 处理玩家的异步回调
            pulseHandlerHumanAsyncEvent();
            // 处理rpc调用返回
            pulseListenRpcResult();
            // 检测rpc超时
            pulseListenRpcTimeout();
            // 处理发送队列中的消息
            pulseSender();
            // 检测玩家掉线，进行数据清理
            pulseHandlerHumanClear();
            // 玩家数据定时存库
            pulseHandlerDbSave();
            // 系统心跳
            GameSystem.pulse();
        } catch (Exception e) {
            LogCore.Server.error("DispatchThread pulse, error : ", e);
        }
    }

    @Override
    protected void pulseHandlerOne(Object data) {
        var message = MessageUtils.fromMessage((byte[]) data, GameMessage.class);
        try {
            var msg = TopicProto.MBasePacketData.parseFrom(message.getBytes());
            var method = ProtoParserUtils.getProtoParserClass(msg.getPacketType().getNumber(), msg.getPacketId());
            Object msgData = null;
            if (method != null) {
                msgData = method.invoke(null, msg.getPacketData());
            }

            switch (msg.getPacketType()) {
                case TopicProto.TOPIC.TOPIC_TYPE_LOGIN: {
                    LoginMsgHandler.handlerLogin(message.getConnectionId(), msg.getPacketId(), msgData);
                    LogCore.GameServer.debug("recv msg, messageId = { {} }, connectionId = {}, packetType = {}, packetId = {}, msgData = {{}}", message.getMessageId(), message.getConnectionId(), msg.getPacketType(), msg.getPacketId(), msgData == null ? "" : msgData.toString().replace("\n", ""));
                    break;
                }
                default:
                    String humanId = HumanObjectManger.humanIds.get(message.getConnectionId());
                    if (humanId == null) {
                        LogCore.GameServer.error("recv msg, messageId = { {} }, connectionId = {}, packetType = {}, packetId = {}, humanObject not found", message.getMessageId(), message.getConnectionId(), msg.getPacketType(), msg.getPacketId());
                        return;
                    }
                    HumanObject humanObject = HumanObjectManger.getHumanObject(humanId);
                    if (humanObject == null) {
                        LogCore.GameServer.error("recv msg, messageId = { {} }, humanId = {}, packetType = {}, packetId = {}, humanObject not found", message.getMessageId(), humanId, msg.getPacketType(), msg.getPacketId());
                        return;
                    }
                    LogCore.GameServer.debug("recv msg, messageId = { {} }, humanId = {}, packetType = {}, packetId = {}, msgData = {{}}", message.getMessageId(), humanId, msg.getPacketType(), msg.getPacketId(), msgData == null ? "" : msgData.toString().replace("\n", ""));
                    // 登录后的消息，添加到玩家消息队列
                    humanObject.getMsgQueue().add(msg);
            }
        } catch (Exception e) {
            LogCore.GameServer.error("recv msg, messageId = { {} }, Exception = {}", message.getMessageId(), e.getLocalizedMessage());
        }
    }

    /**
     * 处理玩家发来的协议
     */
    private void pulseHandlerHumanMsg() {
        for (HumanObject humanObject : HumanObjectManger.getHumanObjects()) {
            while (!humanObject.getMsgQueue().isEmpty()) {
                if (humanObject.isRpcLock()) {
                    break;
                }
                TopicProto.MBasePacketData data = humanObject.getMsgQueue().poll();
                if (data == null) {
                    continue;
                }
                LogicUtils.handler(humanObject, data.getPacketTypeValue(), data.getPacketId(), data.getPacketData());
            }
        }
    }

    /**
     * 处理玩家的异步回调
     */
    private void pulseHandlerHumanAsyncEvent() {
        while (!AsyncEventManager.asyncQueue.isEmpty()) {
            Runnable task = AsyncEventManager.asyncQueue.poll();
            if (task == null) {
                continue;
            }
            task.run();
        }
    }

    /**
     * 检测玩家掉线，进行数据清理
     */
    private void pulseHandlerHumanClear() {
        long cur = System.currentTimeMillis();
        List<String> deleteHumanIds = null;
        for (HumanObject humanObject : HumanObjectManger.getHumanObjects()) {
            if (humanObject.getLastPingTime() + 60 * 1000 < cur) {
                String humanId = humanObject.getHumanId();
                long connectId = humanObject.getConnectObject().getConnectId();
                String uid = humanObject.getConnectObject().getUid();

                humanObject.save();
                DbManager.getDbService().executeAsync("update `human_info` set `role_data` = ? where `human_id` = ?", JSON.toJSONBytes(humanObject.save()), humanId);
                humanObject.getModule(MapModule.class).leaveMap();
                humanObject.kick("ping timeout");

                HumanObjectManger.removeConnectObject(connectId);
                HumanObjectManger.uidAccounts.remove(uid);
                HumanObjectManger.uidPlays.remove(uid);

                if (deleteHumanIds == null) {
                    deleteHumanIds = new ArrayList<>();
                }
                deleteHumanIds.add(humanId);

                List<Long> connectIds = new ArrayList<>();
                for (Map.Entry<Long, String> entry : HumanObjectManger.humanIds.entrySet()) {
                    if (entry.getValue().equals(humanId)) {
                        connectIds.add(entry.getKey());
                    }
                }
                for (Long id : connectIds) {
                    HumanObjectManger.humanIds.remove(id);
                }
                LogCore.GameServer.info("humanId = { {} }, uid = { {} }, clear data", humanId, uid);
            }
        }
        if (deleteHumanIds != null) {
            for (String deleteHumanId : deleteHumanIds) {
                HumanObjectManger.removeHumanObject(deleteHumanId);
            }
        }
    }

    /**
     * 玩家数据定时存库
     */
    private void pulseHandlerDbSave() {
        long cur = System.currentTimeMillis();
        for (HumanObject humanObject : HumanObjectManger.getHumanObjects()) {
            if (humanObject.getLastSaveDbTime() + ToolsUtils.MINUTE_MILLS < cur) {
                humanObject.setLastSaveDbTime(cur);
                DbManager.getDbService().executeAsync("update `human_info` set `role_data` = ? where `human_id` = ?", JSON.toJSONBytes(humanObject.save()),  humanObject.getHumanId());
            }
        }
    }

    /**
     * rpc调用返回处理
     */
    private void pulseListenRpcResult() {
        while (!HumanObjectManger.callBackQueue.isEmpty()) {
            Object data = HumanObjectManger.callBackQueue.poll();
            if (data == null) {
                continue;
            }
            var message = MessageUtils.fromMessage((byte[]) data, Call.class);
            if (message.getType() == CallType.Call.ordinal()) {
                LogCore.RpcClient.debug("rpc result, from NodeId = { {} }, messageId = { {} }, rpcId = {}, result = { {} }, data = {{}}", message.getNodeId(), message.getMessageId(), message.getRpcId(), message.getResult(), message.getData());
                RpcManager.callResult(message);
            } else if (message.getType() == CallType.Update.ordinal()) {
                LogCore.RpcClient.debug("rpc update, cur NodeId = { {} }, from NodeId = { {} }, messageId = { {} }, rpcId = {}, data = {{}}", getNodeId(), message.getNodeId(),  message.getMessageId(), message.getRpcId(), message.getMsg());
                RpcFunction.onUpdate(message);
            }
        }
    }

    /**
     * rpc超时检测
     */
    private void pulseListenRpcTimeout() {
        long cur = System.currentTimeMillis();
        for (Map.Entry<Long, Long> entry : RpcManager.checkTimeout.entrySet()) {
            if (cur > entry.getValue()) {
                RpcManager.checkTimeout.remove(entry.getKey());
                RpcManager.callTimeOut(entry.getKey());
            }
        }
    }
}
